package com.esreport.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.esreport.Utils.DateUtil;
import com.esreport.dao.ReportRepository;
import com.esreport.dto.Action;
import com.esreport.entity.Report;
import com.esreport.service.ReportService;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AbstractAggregationBuilder;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.filter.InternalFilter;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.ExtendedBounds;
import org.elasticsearch.search.aggregations.bucket.histogram.InternalDateHistogram;
import org.elasticsearch.search.aggregations.metrics.cardinality.CardinalityAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.cardinality.InternalCardinality;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.joda.time.DateTimeZone;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.ResultsExtractor;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

@Service
public class ReportServiceImpl implements ReportService {

    @Autowired
    private ReportRepository reportRepository;

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Override
    public long countHits(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long countNewUser(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("action", Action.FIRST_START))
                .must(QueryBuilders.matchQuery("siteId", siteId));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long countBrowse(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("action", Action.VISIT_PAGE)).must(QueryBuilders.matchQuery("nowScene", Action.NEWS_DETAIL))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long sumStayTime(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()));
        //聚合查询 停留时间求和
        SumAggregationBuilder sumBulider = AggregationBuilders.sum("sum_time").field("duration");
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(sumBulider).build();

        double saleAmount = elasticsearchTemplate.query(sq, response -> {
            InternalSum sum = (InternalSum) response.getAggregations().asList().get(0);
            return sum.getValue();
        });
        return (long) saleAmount;
    }

    @Override
    public long sumStayTime(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()));
        //聚合查询 停留时间求和
        SumAggregationBuilder sumBulider = AggregationBuilders.sum("sum_time").field("duration");
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(sumBulider).build();

        double saleAmount = elasticsearchTemplate.query(sq, response -> {
            InternalSum sum = (InternalSum) response.getAggregations().asList().get(0);
            return sum.getValue();
        });
        return (long) saleAmount;
    }

    @Override
    public long countHits(Integer siteId) {
        return reportRepository.countBySiteId(siteId);
    }

    @Override
    public long countShare(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.SHARE));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }


    @Override
    public long countFirstStart(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.FIRST_START));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }


    @Override
    public Long countFirstStartUp(String startTime, String endTime, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(startTime).lte(endTime))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.FIRST_START));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }


    @Override
    public Long countFitstStart(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.FIRST_START));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long distinctCountBrowseNewsUser(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()));

        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public List<Long> groupStartUpCount(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.STARTUP));

        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_hour").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.HOUR).format("yyyy-MM-dd HH").minDocCount(0)
                .minDocCount(0)
                .extendedBounds(new ExtendedBounds(start.getTime(),end.getTime()))
                .timeZone(DateTimeZone.forID("Asia/Shanghai"));
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalDateHistogram agg =  aggregations.get("group_by_hour");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        List<Long> countlist = new ArrayList<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            Long docCount = bucket.getDocCount();
            countlist.add(docCount);
        }

        return countlist;
    }


    @Override
    public long distinctCountFirstStart(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId.toString()));

        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long count = agg.getValue();
        return count;
    }

    @Override
    public long countStartUp(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.STARTUP));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long countStartUpForToday(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(DateUtil.getDayBegin().getTime())
                .lte(DateUtil.getDayEnd().getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.STARTUP));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long countSeriesSign(Integer siteId, Date start, Date end) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId",siteId.toString()))
                .must(QueryBuilders.rangeQuery("startTime").gte(DateUtil.getDayBegin().getTime())
                        .lte(DateUtil.getDayEnd().getTime()));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("SeriesSignCount").field("seriesSign").precisionThreshold(40000);
        CardinalityAggregationBuilder car1 = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).addAggregation(car1).build();
        return elasticsearchTemplate.count(sq,Report.class);
    }

    @Override
    public List<Long> groupNewUserCount(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", Action.FIRST_START));
        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_day").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .extendedBounds(new ExtendedBounds(start.getTime(),end.getTime()))
                .timeZone(DateTimeZone.forID("Asia/Shanghai"));
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalDateHistogram agg =  aggregations.get("group_by_day");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        List<Long> countlist = new ArrayList<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            Long docCount = bucket.getDocCount();
            String date = bucket.getKeyAsString();
            countlist.add(docCount);
        }
        return countlist;
    }

    @Override
    public Long getPhoneTypeCount(Date end, Integer siteId,String phoneType) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("phoneType", phoneType))
                .must(QueryBuilders.rangeQuery("startTime").lte(end.getTime()));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        return elasticsearchTemplate.count(sq, Report.class);

    }

    @Override
    public Long getInUserMemberCount(Date start,Date end, Integer siteId, String[] deviceIdList) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
//                .must(QueryBuilders.matchQuery("action", Action.STARTUP))
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("deviceId",deviceIdList));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        return elasticsearchTemplate.count(sq,Report.class);
    }

    @Override
    public String[] getFirstStartUserList(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("action",Action.FIRST_START));

        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).withPageable(PageRequest.of(0,1000)).build();
        List<Report> list = elasticsearchTemplate.queryForList(sq,Report.class);
        List<String> deviceIdList = new ArrayList<>();
        if(list!=null&&list.size()>0){
            for (Report report:list) {
                String deviceId = report.getDeviceId();
                deviceIdList.add(deviceId);
            }
        }
        return deviceIdList.toArray(new String[deviceIdList.size()]);
    }

    @Override
    public Long getActionCount(Date start, Date end, Integer siteId, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId",siteId.toString()))
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("action",action));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq,Report.class);
    }

    @Override
    public Map<String,Long> groupDayForActionByAction(Date start, Date end, Integer siteId, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", action));
        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_day").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .extendedBounds(new ExtendedBounds(start.getTime(),end.getTime()))
                .timeZone(DateTimeZone.forID("Asia/Shanghai"));
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, SearchResponse::getAggregations);
        InternalDateHistogram agg =  aggregations.get("group_by_day");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        Map<String,Long> countMap = new LinkedHashMap<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            Long docCount = bucket.getDocCount();
            String date = bucket.getKeyAsString();
            countMap.put(date,docCount);
        }
        return countMap;
    }

    @Override
    public Map<String,Long> groupHourForActionByAction(Date start, Date end, Integer siteId, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", action));
        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_hour").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.HOUR).format("yyyy-MM-dd HH").minDocCount(0)
                .extendedBounds(new ExtendedBounds(start.getTime(),end.getTime()))
                .timeZone(DateTimeZone.forID("Asia/Shanghai"));
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, SearchResponse::getAggregations);
        InternalDateHistogram agg =  aggregations.get("group_by_hour");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        Map<String,Long> countMap = new LinkedHashMap<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            Long docCount = bucket.getDocCount();
            String date = bucket.getKeyAsString();
            countMap.put(date,docCount);
        }
        return countMap;
    }

    @Override
    public long distinctCountFirstStartForHuanCui(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action",Action.FIRST_START));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public long countHitsDistinct(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();

        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long UV = agg.getValue();
        return UV;
    }

    @Override
    public long countHitsDistinct(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long UV = agg.getValue();
        return UV;
    }

    @Override
    public long countHitsDistinctIP(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("ip").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long IP = agg.getValue();
        return IP;
    }

    @Override
    public long countHitsDistinctIP(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.matchQuery("siteId", siteId));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("ip").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long IP = agg.getValue();
        return IP;
    }

    @Override
    public Long countStartUp(Date start, Date end, Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("siteId", siteId)).must(QueryBuilders.matchQuery("action",Action.STARTUP));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long startUpCount = agg.getValue();
        return startUpCount;
    }

    @Override
    public Long distinctCountStartUp(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId)).must(QueryBuilders.matchQuery("action",Action.STARTUP));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long startUpCount = agg.getValue();
        return startUpCount;
    }

    @Override
    public Long countNum(Integer siteId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long startUpCount = agg.getValue();
        return startUpCount;
    }

    @Override
    public Long countNum(Integer siteId, Date start, Date end) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId))
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long count = agg.getValue();
        return count;
    }

    @Override
    public Long countFirstStartBetweenDateNum(Integer siteId, String start, String end, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId))
                .must(QueryBuilders.matchQuery("action",action))
                .must(QueryBuilders.rangeQuery("startTime").gte(start).lte(end));
        CardinalityAggregationBuilder car = AggregationBuilders.cardinality("deviceCount").field("deviceId").precisionThreshold(40000);
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).addAggregation(car).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalCardinality agg = aggregations.get("deviceCount");
        Long count = agg.getValue();
        return count;
    }

    @Override
    public List<JSONObject> countFirstStartGroupByDay(Integer siteId, String start, String end, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start).lte(end))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", action));

        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_hour").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .minDocCount(0)
                .extendedBounds(new ExtendedBounds(start,end))
                .timeZone(DateTimeZone.forID("Asia/Shanghai"));
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalDateHistogram agg =  aggregations.get("group_by_hour");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        List<JSONObject> countlist = new ArrayList<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            JSONObject obj = new JSONObject();
            Long docCount = bucket.getDocCount();
            obj.put("date",bucket.getKeyAsString());
            obj.put("docCount",docCount);
            countlist.add(obj);
        }

        return countlist;
    }

    @Override
    public Long countByAction(Integer siteId, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("siteId", siteId.toString()))
                .must(QueryBuilders.matchQuery("action", action));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public List<JSONObject> groupCountDangShiStudyByDay(Integer siteId, String start, String end) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("startTime").gte(start).lte(end))
                .must(QueryBuilders.matchQuery("action", Action.DANG_SHI_STUDY))
                .must(QueryBuilders.matchQuery("siteId", siteId.toString()));

        AbstractAggregationBuilder subAggregation = AggregationBuilders.cardinality("car_username").field("userName").precisionThreshold(40000);

        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_day").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .extendedBounds(new ExtendedBounds(start,end))
                .timeZone(DateTimeZone.forID("Asia/Shanghai")).subAggregation(subAggregation);
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalDateHistogram agg =  aggregations.get("group_by_day");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        List<JSONObject> countlist = new ArrayList<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            JSONObject obj = new JSONObject();
            Long docCount = bucket.getDocCount();
            obj.put("date",bucket.getKeyAsString());
            obj.put("docCount",docCount);
            Map<String,Aggregation> map = bucket.getAggregations().getAsMap();
            String text = map.get("car_username").toString();
            JSONObject json = JSON.parseObject(text);
            json = json.getJSONObject("car_username");
            Long value = json.getLong("value");
            obj.put("disPeopleCount",value);
            countlist.add(obj);
        }

        return countlist;
    }

    @Override
    public Long countByAction(String username, String action) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("userName", username))
                .must(QueryBuilders.matchQuery("action", action));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public Long countByActionAndShop(String action, Integer shopId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("shopId", shopId))
                .must(QueryBuilders.matchQuery("action", action));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq,Report.class);
    }

    @Override
    public Long countByActionAndShop(String action, Integer shopId, Date beginTime, Date endTime) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("shopId", shopId))
                .must(QueryBuilders.matchQuery("action", action))
                .must(QueryBuilders.rangeQuery("startTime").gte(beginTime.getTime()))
                .must(QueryBuilders.rangeQuery("startTime").lte(endTime.getTime()));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return  elasticsearchTemplate.count(sq,Report.class);
    }

    @Override
    public Long countByActionAndAd(String action, Integer advertisePutId) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("advertisePutId", advertisePutId))
                .must(QueryBuilders.matchQuery("action", action));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq,Report.class);
    }

    @Override
    public Long countByActionAndAd(String action, Integer advertisePutId, Date beginTime, Date endTime) {
        QueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("advertisePutId", advertisePutId))
                .must(QueryBuilders.matchQuery("action", action))
                .must(QueryBuilders.rangeQuery("startTime").gte(beginTime.getTime()))
                .must(QueryBuilders.rangeQuery("startTime").lte(endTime.getTime()));
        SearchQuery sq = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
        return elasticsearchTemplate.count(sq, Report.class);
    }

    @Override
    public  List<Map<String,Object>> getAdDayHistogram(Date start, Date end, Integer advertisePutId) {

        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(QueryBuilders.rangeQuery("startTime").gte(start.getTime()).lte(end.getTime()))
                .must(QueryBuilders.matchQuery("advertisePutId",advertisePutId));


        AbstractAggregationBuilder browseAdFilter = AggregationBuilders.filter("browseAdvertiseFilter", QueryBuilders.termQuery("action","browseAdvertise"));
                browseAdFilter.subAggregation(AggregationBuilders.dateHistogram("group_by_day").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .extendedBounds(new ExtendedBounds(start.getTime(),end.getTime()))
                .timeZone(DateTimeZone.forID("Asia/Shanghai")));

        AbstractAggregationBuilder clickAdFilter = AggregationBuilders.filter("clickAdvertiseFilter", QueryBuilders.termQuery("action","clickAdvertise"));
        clickAdFilter.subAggregation(AggregationBuilders.dateHistogram("group_by_day").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .extendedBounds(new ExtendedBounds(start.getTime(),end.getTime()))
                .timeZone(DateTimeZone.forID("Asia/Shanghai")));

        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(browseAdFilter).addAggregation(clickAdFilter).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, SearchResponse::getAggregations);

        InternalFilter browseAdvertiseFilter = aggregations.get("browseAdvertiseFilter");
        InternalDateHistogram browseGroupByDay =browseAdvertiseFilter.getAggregations().get("group_by_day");
        List<InternalDateHistogram.Bucket> browseBuckets = browseGroupByDay.getBuckets();
        List<Map<String,Object>> dateCountList = new ArrayList();

        for (InternalDateHistogram.Bucket bucket:browseBuckets) {
            Long docCount = bucket.getDocCount();
            String date = bucket.getKeyAsString();
            Map<String,Object> dataMap =new HashMap();
            dataMap.put("date",date);
            dataMap.put("browse",docCount);
            dataMap.put("click",0);
            dateCountList.add(dataMap);
        }


        InternalFilter clickAdvertiseFilter = aggregations.get("clickAdvertiseFilter");
        InternalDateHistogram clickGroupByDay =clickAdvertiseFilter.getAggregations().get("group_by_day");
        List<InternalDateHistogram.Bucket> clickBuckets = clickGroupByDay.getBuckets();
        for (InternalDateHistogram.Bucket bucket:clickBuckets) {
            Long docCount = bucket.getDocCount();
            String date = bucket.getKeyAsString();
            Map<String,Object> dataMap = dateCountList.stream().filter(m->m.get("date").equals(date)).findFirst().orElse(new HashMap());
            if(dataMap.isEmpty()){
                dataMap.put("browse",0L);
                dataMap.put("date",date);
                dateCountList.add(dataMap);
            }
            dataMap.put("click",docCount);
        }
        return dateCountList;
    }

    @Override
    public List<JSONObject> groupDangShiStudyByUsername(String username, String start, String end, String action) {
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("action", action))
                .must(QueryBuilders.matchQuery("userName", username));
            if(StringUtils.isNotBlank(start)){
                queryBuilder.must(QueryBuilders.rangeQuery("startTime").gte(start));
            }
            if(StringUtils.isNotBlank(end)){
                queryBuilder.must(QueryBuilders.rangeQuery("startTime").lte(end));
            }
        AbstractAggregationBuilder aggregation = AggregationBuilders.dateHistogram("group_by_day").field("startTime")
                .dateHistogramInterval(DateHistogramInterval.DAY).format("yyyy-MM-dd").minDocCount(0)
                .timeZone(DateTimeZone.forID("Asia/Shanghai"));
        SearchQuery sq = new NativeSearchQueryBuilder().withIndices("esreport2").withTypes("report2").withQuery(queryBuilder)
                .addAggregation(aggregation).build();
        Aggregations aggregations = elasticsearchTemplate.query(sq, new ResultsExtractor<Aggregations>() {
            @Override
            public Aggregations extract(SearchResponse searchResponse) {
                return searchResponse.getAggregations();
            }
        });
        InternalDateHistogram agg =  aggregations.get("group_by_day");
        List<InternalDateHistogram.Bucket> buckets = agg.getBuckets();
        List<JSONObject> countlist = new ArrayList<>();
        for (InternalDateHistogram.Bucket bucket:buckets) {
            JSONObject obj = new JSONObject();
            Long docCount = bucket.getDocCount();
            String date = bucket.getKeyAsString();
            obj.put("date",date);
            obj.put("docCount",docCount);
            countlist.add(obj);
        }
        return countlist;
    }
}
